/**
 * 
 */
package spark.yk.java.maven;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.rdd.RDD;

import com.google.common.base.Predicate;

import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

/**
 * @author 林智星
 *
 */
public class SimpleApp {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		
		String uri = "hdfs://192.168.1.185:9000";  
	    String dir = "/user/output1";  
	    String parentDir = "/user";  
	    String remoteFile = uri+dir + "/README.md";
		// TODO Auto-generated method stub
		String logFile = remoteFile; // Should be some file on your system
		SparkConf conf = new SparkConf().setAppName("Simple Application").setMaster("spark://192.168.1.185:7077");
		SparkContext sc = new SparkContext(conf);
		JavaSparkContext ctx = new JavaSparkContext(sc);
		ctx.addJar("D:\\Java\\workspace\\SparkJavaDemo\\SimpleApp.jar");
		
	    //JavaRDD<String> logData = sc.textFile(logFile, 2).cache().toJavaRDD();
		//long numAs = logData.filter(line -> line.contains("a")).count();
	     //long numBs = logData.filter(line -> line.contains("b")).count();
		JavaRDD<String> lines = ctx.textFile(logFile, 2);
		long numAs = lines.filter(line->line.contains("a")).count();
		long numBs = lines.filter(line->line.contains("b")).count();
		
		/*long numAs = lines.filter(
				
				new Function<String, Boolean>() {
		            public Boolean call(String v1) throws Exception {
		                return v1.contains("a");
		            }
		        }
		).count();*/
		
		/*
	     RDD<String> logData = sc.textFile(logFile, 2).cache();
	     Predicate<String> containA = line -> line.contains("a"); 
	     Predicate<String> containB = line -> line.contains("b");
	     long numAs = logData.filter(new Function1<String, String>() {
	          public String call(String s) { 
	              System.out.println("每行长度"+s.length());
	              if(s.contains("a"))
	               return s; }
	        }).count();
	     long numBs = logData.filter(line -> line.contains("b")).count();
	     */
		//System.out.println("Lines with a: $numAs");
	     System.out.println("Lines with a: $numAs, Lines with b: $numBs");
		 sc.stop();
	}

}
